package cn.com.duiba.millionaire.center.biz.service.impl;

import cn.com.duiba.millionaire.center.api.constant.RedisKeyFactory;
import cn.com.duiba.millionaire.center.api.dto.ConReLiveCardSearchDto;
import cn.com.duiba.millionaire.center.api.dto.LoadReliveCardDto;
import cn.com.duiba.millionaire.center.api.dto.MillionaireConsumerDto;
import cn.com.duiba.millionaire.center.api.dto.UserReLiveCardParamDto;
import cn.com.duiba.millionaire.center.api.enums.RenascenceTypeEnum;
import cn.com.duiba.millionaire.center.api.msg.LoadReliveCardMsg;
import cn.com.duiba.millionaire.center.biz.bo.ConsumeRenascenceParam;
import cn.com.duiba.millionaire.center.biz.bo.ConsumerRenascenceInfoParam;
import cn.com.duiba.millionaire.center.biz.bo.UserReLiveCardParam;
import cn.com.duiba.millionaire.center.biz.dao.MillionaireConsumerDao;
import cn.com.duiba.millionaire.center.biz.dao.MillionaireRenascenceRecordDao;
import cn.com.duiba.millionaire.center.biz.dao.TableHelper;
import cn.com.duiba.millionaire.center.biz.entity.MillionaireConsumerEntity;
import cn.com.duiba.millionaire.center.biz.entity.MillionaireRenascenceRecordEntity;
import cn.com.duiba.millionaire.center.biz.service.RenascenceRecordService;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * Created by sty on 1/23/18.
 */
@Service
public class RenascenceRecordServiceImpl implements RenascenceRecordService {

    @Resource(name = "redisTemplate")
    private RedisTemplate<String,Long> redisTemplate;
    @Autowired
    private MillionaireConsumerDao millionaireConsumerDao;
    @Autowired
    private MillionaireRenascenceRecordDao millionaireRenascenceRecordDao;
    @Autowired
    private ExecutorService executorService;


    @Override
    public void consumeRenascenceCard(List<ConsumeRenascenceParam> consumeRenascenceParams) {

        if (CollectionUtils.isEmpty(consumeRenascenceParams)) {
            return;
        }

        checkParam(consumeRenascenceParams);

        List<ConsumerRenascenceInfoParam> totalInfo = Lists.newArrayList();
        for (ConsumeRenascenceParam one : consumeRenascenceParams){
            totalInfo.addAll( one.getCid().stream().map(user->{
                ConsumerRenascenceInfoParam param = new ConsumerRenascenceInfoParam();
                param.setActivityId(one.getActivityId());
                param.setNum(one.getNum());
                param.setQuestionIndex(one.getQuestionIndex());
                param.setCid(user);
                return param;
            }).collect(Collectors.toList()));
         }

        //同步消耗用户redis的复活卡
        List<ConsumerRenascenceInfoParam> restUsers=Lists.newArrayList();
        HashOperations<String, Long, Integer> hashOperations = redisTemplate.opsForHash();

        //根据活动id进行批量查找redis----目前只有这个办法
        Map<Long, List<ConsumerRenascenceInfoParam>> aidPrefix = Maps.newHashMap();
        totalInfo.stream().forEach(one -> {
            List<ConsumerRenascenceInfoParam> origin = aidPrefix.get(one.getActivityId());
            if (origin == null) {
                aidPrefix.put(one.getActivityId(), Lists.newArrayList(one));
            } else {
                origin.add(one);
                aidPrefix.put(one.getActivityId(), origin);
            }
        });

        Map<Long,ConsumerRenascenceInfoParam> userRenascenceMap=Maps.newHashMap();

        for(Map.Entry<Long, List<ConsumerRenascenceInfoParam>> entry : aidPrefix.entrySet()){
            if(CollectionUtils.isEmpty(entry.getValue())){
                continue;
            }
           List<Integer> cardInRedis = hashOperations.multiGet(RedisKeyFactory.K012.toString()+entry.getKey(), totalInfo.stream().map(ConsumerRenascenceInfoParam::getCid).collect(Collectors.toList()));
          for(int i=0;i<entry.getValue().size();i++){
               ConsumerRenascenceInfoParam consumer  = entry.getValue().get(i);
               Integer userCard = cardInRedis.get(i);
               if(userCard==null){
                   restUsers.add(consumer);
                   userRenascenceMap.put(consumer.getCid(),consumer);
                   continue;
               }
               consumer.setOriginRedisCardNum(cardInRedis.get(i));
               userRenascenceMap.put(consumer.getCid(),consumer);
           }

        }

        //批量查库导出
        Map<String, List<ConsumerRenascenceInfoParam>> userPrefix = Maps.newHashMap();
        restUsers.stream().forEach(one -> {
            String prefix = TableHelper.getTableSuffix(one.getCid());
            List<ConsumerRenascenceInfoParam> origin = userPrefix.get(prefix);
            if (origin == null) {
                userPrefix.put(prefix, Lists.newArrayList(one));
            } else {
                origin.add(one);
                userPrefix.put(prefix, origin);
            }
        });
        List<MillionaireConsumerEntity> consumers= Lists.newArrayList();
        for (Map.Entry<String, List<ConsumerRenascenceInfoParam>> entry : userPrefix.entrySet()) {
            consumers.addAll( millionaireConsumerDao.batchFindConsumer(entry.getValue().stream().map(ConsumerRenascenceInfoParam::getCid).sorted().collect(Collectors.toList()), entry.getKey()));
        }
        consumers.stream().forEach(one->{
            ConsumerRenascenceInfoParam dbConsumer =  userRenascenceMap.get(one.getConsumerId());
            dbConsumer.setOriginRedisCardNum(one.getRenascenceNum());
        });
        //放进redis 操作

        hashOperations.getOperations().executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer<Object> keyS = (RedisSerializer<Object>) hashOperations.getOperations().getKeySerializer();
                RedisSerializer<Object> hashKeyS = (RedisSerializer<Object>) hashOperations.getOperations().getHashKeySerializer();
                RedisSerializer<Object> valueKeyS = (RedisSerializer<Object>) hashOperations.getOperations().getHashValueSerializer();
                for (Map.Entry<Long,ConsumerRenascenceInfoParam> one: userRenascenceMap.entrySet()) {

                    ConsumerRenascenceInfoParam user = one.getValue();
                    byte[] keyByteK012 = keyS.serialize(RedisKeyFactory.K012.toString()+user.getActivityId());
                    Integer originReNum = user.getOriginRedisCardNum();
                    if(originReNum==null){
                        continue;
                    }
                    connection.hSet(keyByteK012, hashKeyS.serialize(user.getCid()), valueKeyS.serialize(originReNum-user.getNum()));
                    connection.expire(keyByteK012, 3600);
                }
                return null;
            }
        });

        //异步同步数据库扣减
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                Map<String, List<ConsumerRenascenceInfoParam>> userPrefixTotal = Maps.newHashMap();

                totalInfo.stream().forEach(one -> {
                    String prefix = TableHelper.getTableSuffix(one.getCid());
                    ConsumerRenascenceInfoParam param = new ConsumerRenascenceInfoParam();
                    param.setCid(one.getCid());
                    param.setNum(one.getNum());
                    param.setActivityId(one.getActivityId());
                    List<ConsumerRenascenceInfoParam> origin = userPrefixTotal.get(prefix);
                    if (origin == null) {
                        userPrefixTotal.put(prefix, Lists.newArrayList(param));
                    } else {
                        origin.add(param);
                        userPrefixTotal.put(prefix, origin);
                    }
                });

                for (Map.Entry<String, List<ConsumerRenascenceInfoParam>> entry : userPrefixTotal.entrySet()) {

                    millionaireConsumerDao.batchResumeConsumerRenascence(entry.getValue().stream().sorted().collect(Collectors.toList()), entry.getKey());

                }

                redisTemplate.delete(totalInfo.stream().map(one->RedisKeyFactory.K008.toString()+one.getCid()).collect(Collectors.toList()));
            }
        });
        //异步增加复活卡使用记录
        executorService.submit(new Runnable() {
            @Override
            public void run() {
                List<MillionaireRenascenceRecordEntity> records = totalInfo.stream()
                        .map(one -> {
                            MillionaireRenascenceRecordEntity record = new MillionaireRenascenceRecordEntity();
                            record.setConsumerId(one.getCid());
                            record.setCount(one.getNum());
                            record.setBizType(RenascenceTypeEnum.RENASCENCE.getCode());
                            record.setBizNum("activityId:"+one.getActivityId()+"_"+"index:"+one.getQuestionIndex());
                            return record;
                        }).collect(Collectors.toList());
                millionaireRenascenceRecordDao.batchInsert(records);
            }
        });
    }

    @Override
    public void batchLoadFromDatabase(List<LoadReliveCardMsg> msgs) {
        if(CollectionUtils.isEmpty(msgs)){
            return;
        }
        //用户1024分表
        Map<String, List<LoadReliveCardMsg>> userPrefix = Maps.newHashMap();

        Map<Long,Long> userMapActivityId = Maps.newHashMap();
        msgs.stream().forEach(one -> {
            String prefix = TableHelper.getTableSuffix(one.getConsumerId());
            List<LoadReliveCardMsg> origin = userPrefix.get(prefix);
            if (origin == null) {
                userPrefix.put(prefix, Lists.newArrayList(one));
            } else {
                origin.add(one);
                userPrefix.put(prefix, origin);
            }
            userMapActivityId.put(one.getConsumerId(),one.getActivityId());
        });

        List<MillionaireConsumerEntity> consumers= Lists.newArrayList();

        for (Map.Entry<String, List<LoadReliveCardMsg>> entry : userPrefix.entrySet()) {
            consumers.addAll( millionaireConsumerDao.batchFindConsumer(entry.getValue().stream().map(LoadReliveCardMsg::getConsumerId).sorted().collect(Collectors.toList()), entry.getKey()));
        }

        List<LoadReliveCardDto> reliveCardInfo  = Lists.newArrayList();
        consumers.stream().forEach(one->{
            Long activityId = userMapActivityId.get(one.getConsumerId());
            LoadReliveCardDto unit = new LoadReliveCardDto();
            unit.setActivityId(activityId);
            unit.setConsumerId(one.getConsumerId());
            unit.setReliveCardNum(one.getRenascenceNum()==null?0:one.getRenascenceNum());
            reliveCardInfo.add(unit);
        });

        HashOperations<String, Long, Integer> hashOperations = redisTemplate.opsForHash();
        hashOperations.getOperations().executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer<Object> keyS = (RedisSerializer<Object>) hashOperations.getOperations().getKeySerializer();
                RedisSerializer<Object> hashKeyS = (RedisSerializer<Object>) hashOperations.getOperations().getHashKeySerializer();
                RedisSerializer<Object> hashValueS = (RedisSerializer<Object>) hashOperations.getOperations().getHashValueSerializer();

                for (LoadReliveCardDto one: reliveCardInfo) {
                    byte[] keyByte = keyS.serialize(RedisKeyFactory.K012.toString()+one.getActivityId());
                    connection.hSet(keyByte, hashKeyS.serialize(one.getConsumerId()) ,hashValueS.serialize(one.getReliveCardNum()));
                    connection.expire(keyByte, 3600);
                }
                return null;
            }
        });

    }

    @Override
    public Map<Long,Integer> findConsumerReLiveCardByCids(List<ConReLiveCardSearchDto> users){
        Map<Long,Integer> consumerRenascence = Maps.newHashMap();
        if(CollectionUtils.isEmpty(users)){
            return consumerRenascence;
        }
        checkReLiveParam(users);

        Map<Long, List<ConReLiveCardSearchDto>> aidPrefix = Maps.newHashMap();
        users.stream().forEach(one -> {
            List<ConReLiveCardSearchDto> origin = aidPrefix.get(one.getActivityId());
            if (origin == null) {
                aidPrefix.put(one.getActivityId(), Lists.newArrayList(one));
            } else {
                origin.add(one);
                aidPrefix.put(one.getActivityId(), origin);
            }
        });

        List<ConReLiveCardSearchDto> notInRedis = Lists.newArrayList();
        HashOperations<String, Long, Integer> hashOperations = redisTemplate.opsForHash();
        for(Map.Entry<Long, List<ConReLiveCardSearchDto>> entry : aidPrefix.entrySet()){
            if(CollectionUtils.isEmpty(entry.getValue())){
                continue;
            }

            List<Integer> cardInRedis = hashOperations.multiGet(RedisKeyFactory.K012.toString()+entry.getKey(), entry.getValue().stream().map(ConReLiveCardSearchDto::getConsumerId).collect(Collectors.toList()));
               for(int i=0;i<entry.getValue().size();i++){
                   ConReLiveCardSearchDto rCDto  = entry.getValue().get(i);
                   if(rCDto==null){
                       continue;
                   }
                   Integer num ;
                   if(i<cardInRedis.size()){
                       num = cardInRedis.get(i);
                       if(num!=null){
                           consumerRenascence.put(rCDto.getConsumerId(),num);
                       }else{
                           notInRedis.add(rCDto);
                       }
                   }

            }
        }

        Map<String, List<ConReLiveCardSearchDto>> userPrefix = Maps.newHashMap();
        notInRedis.stream().forEach(one -> {
            String prefix = TableHelper.getTableSuffix(one.getConsumerId());
            List<ConReLiveCardSearchDto> origin = userPrefix.get(prefix);
            if (origin == null) {
                userPrefix.put(prefix, Lists.newArrayList(one));
            } else {
                origin.add(one);
                userPrefix.put(prefix, origin);
            }
        });
        List<MillionaireConsumerEntity> consumers= Lists.newArrayList();
        for (Map.Entry<String, List<ConReLiveCardSearchDto>> entry : userPrefix.entrySet()) {
            consumers.addAll( millionaireConsumerDao.batchFindConsumer(entry.getValue().stream().map(ConReLiveCardSearchDto::getConsumerId).sorted().collect(Collectors.toList()), entry.getKey()));
        }
        Map<Long,Integer> userReLiveNumMap = Maps.newHashMap();
        consumers.stream().forEach(one->{
            userReLiveNumMap.put(one.getConsumerId(),one.getRenascenceNum());
            consumerRenascence.put(one.getConsumerId(),one.getRenascenceNum());
        });
        hashOperations.getOperations().executePipelined(new RedisCallback<Object>() {
            @Override
            public Object doInRedis(RedisConnection connection) throws DataAccessException {
                RedisSerializer<Object> keyS = (RedisSerializer<Object>) hashOperations.getOperations().getKeySerializer();
                RedisSerializer<Object> hashKeyS = (RedisSerializer<Object>) hashOperations.getOperations().getHashKeySerializer();
                RedisSerializer<Object> valueKeyS = (RedisSerializer<Object>) hashOperations.getOperations().getHashValueSerializer();
                for (ConReLiveCardSearchDto dto: notInRedis) {
                    byte[] keyByteK012 = keyS.serialize(RedisKeyFactory.K012.toString()+dto.getActivityId());
                    Integer originReNum = userReLiveNumMap.get(dto.getConsumerId());
                    if(originReNum==null){
                        continue;
                    }
                    connection.hSet(keyByteK012, hashKeyS.serialize(dto.getConsumerId()), valueKeyS.serialize(originReNum));
                    connection.expire(keyByteK012, 3600);
                }
                return null;
            }
        });
        return consumerRenascence;
    }

    @Override
    public boolean batchUpdateConsumerReLiveCard(Long appId, List<UserReLiveCardParamDto> paramList) {
        if(CollectionUtils.isEmpty(paramList) || appId==null){
            return false;
        }
        Map<String, List<UserReLiveCardParamDto>> userPrefix = Maps.newHashMap();
        paramList.stream().forEach(one -> {
            String prefix = TableHelper.getTableSuffix(one.getConsumerId());
            List<UserReLiveCardParamDto> origin = userPrefix.get(prefix);
            if (origin == null) {
                userPrefix.put(prefix, Lists.newArrayList(one));
            } else {
                origin.add(one);
                userPrefix.put(prefix, origin);
            }
        });
        //批量增加复活卡记录
        int ret = millionaireRenascenceRecordDao.batchInsert(paramList.stream().map(param->{
            MillionaireRenascenceRecordEntity record = new MillionaireRenascenceRecordEntity();
            record.setConsumerId(param.getConsumerId());
            record.setCount(param.getChangeNum());
            record.setBizType(RenascenceTypeEnum.DEVELOPAPPLY.getCode());
            record.setBizNum(param.getBizNum());//作为一个是否一批增加的用户标识
            return record;
        }).collect(Collectors.toList()));


        //先查询用户是否在用户库存在
        List<MillionaireConsumerEntity> consumerList = Lists.newArrayList();
        Map<Long,MillionaireConsumerEntity> consumer = Maps.newConcurrentMap();
        for (Map.Entry<String, List<UserReLiveCardParamDto>> entry : userPrefix.entrySet()) {
            consumerList.addAll( millionaireConsumerDao.batchFindConsumer(entry.getValue().stream().map(UserReLiveCardParamDto::getConsumerId).collect(Collectors.toList()), entry.getKey()) );
        }
        consumerList.stream().forEach(one->{
            consumer.put(one.getConsumerId(), one);
        });

        List<UserReLiveCardParamDto> newUser  = paramList.stream().filter(one->!consumer.containsKey(one.getConsumerId())).collect(Collectors.toList());
        if(CollectionUtils.isNotEmpty(newUser)){
            Map<String, List<UserReLiveCardParamDto>> newUserPrefix = Maps.newHashMap();
            newUser.stream().forEach(one -> {
                String prefix = TableHelper.getTableSuffix(one.getConsumerId());
                List<UserReLiveCardParamDto> origin = newUserPrefix.get(prefix);
                if (origin == null) {
                    newUserPrefix.put(prefix, Lists.newArrayList(one));
                } else {
                    origin.add(one);
                    newUserPrefix.put(prefix, origin);
                }
            });
            for (Map.Entry<String, List<UserReLiveCardParamDto>> entry : newUserPrefix.entrySet()) {
                List<MillionaireConsumerEntity> newUserList = entry.getValue().stream().map(one->{
                    MillionaireConsumerEntity newConsumer = new MillionaireConsumerEntity();
                    newConsumer.setAppId(appId);
                    newConsumer.setRenascenceNum(0);
                    newConsumer.setAccount(0L);
                    newConsumer.setConsumerId(one.getConsumerId());
                    return newConsumer;
                }).collect(Collectors.toList());
                millionaireConsumerDao.batchInsert(newUserList, entry.getKey());
            }
        }

        redisTemplate.delete(Lists.newArrayList(paramList.stream().map(one->(RedisKeyFactory.K008.toString()+one.getConsumerId())).collect(Collectors.toList())));

        //批量增加记录
        for (Map.Entry<String, List<UserReLiveCardParamDto>> entry : userPrefix.entrySet()) {
             millionaireConsumerDao.batchUpdateRenascenceNum(entry.getValue().stream().map(one->{
                     UserReLiveCardParam param = new UserReLiveCardParam();
                     param.setChangeNum(one.getChangeNum());
                     param.setConsumerId(one.getConsumerId());
                     return param;
             }).sorted().collect(Collectors.toList()), entry.getKey());
        }

        return true;
    }


    private void checkParam(List<ConsumeRenascenceParam> consumeRenascenceParams) {
        for (ConsumeRenascenceParam param : consumeRenascenceParams) {
            if (param.getNum() < 1) {
                throw new RuntimeException("扣减数量不能小于1");
            }
            if (param.getCid().size() > 1000) {
                throw new RuntimeException("不支持1000个以上的数据");
            }
        }
    }

    private void checkReLiveParam(List<ConReLiveCardSearchDto> users) {

        for (ConReLiveCardSearchDto param : users) {
            if (param.getActivityId() == null) {
                throw new RuntimeException("活动id不能为空");
            }
            if (param.getConsumerId()==null) {
                throw new RuntimeException("用户id不能为空");
            }
        }
    }
}
